package data

import (
	"context"
	"fmt"
	consul "github.com/go-kratos/consul/registry"
	"github.com/go-kratos/kratos/v2/log"
	"github.com/go-kratos/kratos/v2/registry"
	"go_private_im/api/comet"
	"go_private_im/internal/logic/biz"
	"go_private_im/internal/logic/conf"
	"google.golang.org/grpc"
	"google.golang.org/grpc/keepalive"
	"net/url"
	"strings"
	"time"
)

const (
	_prefixMidServer    = "userId_%d" // userId -> key:server
	_prefixKeyServer    = "key_%s" // key -> server
	_prefixServerOnline = "ol_%s"  // server -> online
	_onlineTick         = time.Second * 10
	_onlineDeadline     = time.Minute * 5
)
var (
	// grpc options
	grpcKeepAliveTime    = time.Duration(10) * time.Second
	grpcKeepAliveTimeout = time.Duration(3) * time.Second
	grpcBackoffMaxDelay  = time.Duration(3) * time.Second
	grpcMaxSendMsgSize   = 1 << 24
	grpcMaxCallMsgSize   = 1 << 24
)

const (
	// grpc options
	grpcInitialWindowSize     = 1 << 24
	grpcInitialConnWindowSize = 1 << 24
)
func keyUserIdServer(userId int64) string {
	return fmt.Sprintf(_prefixMidServer, userId)
}

func keyKeyServer(key string) string {
	return fmt.Sprintf(_prefixKeyServer, key)
}

func keyServerOnline(key string) string {
	return fmt.Sprintf(_prefixServerOnline, key)
}

type LogicRepo struct {
	nodes       []*registry.ServiceInstance
	data        *Data
	conf        *conf.Config
	log         *log.Helper
	redisExpire int32
	roomCount   map[string]int32
	CometServers map[string]comet.CometClient
}

// NewGreeterRepo .
func NewLogicRepo(data *Data, logger log.Logger, conf *conf.Config) biz.LogicRepo {
	l := &LogicRepo{
		log:         log.NewHelper(logger),
		data:        data,
		conf:        conf,
		redisExpire: int32(conf.Redis.Expire / time.Second),
	}
	go l.newNodes()
	go l.onlineproc()
	return l
}

func (l *LogicRepo) onlineproc() {
	for {
		time.Sleep(_onlineTick)
		if err := l.loadOnline(); err != nil {
			l.log.Errorf("onlineproc error(%v)", err)
		}
	}
}
func (l *LogicRepo) loadOnline() (err error) {
	var (
		roomCount = make(map[string]int32)
	)
	for _, server := range l.nodes {
		var online *biz.Online
		online, err = l.ServerOnline(context.Background(), server.ID)
		if err != nil {
			return
		}
		if time.Since(time.Unix(online.Updated, 0)) > _onlineDeadline {
			_ = l.DelServerOnline(context.Background(), server.ID)
			continue
		}
		for roomId, count := range online.RoomCount {
			roomCount[roomId] += count
		}
	}
	l.roomCount = roomCount
	return
}

func (l *LogicRepo) newNodes() {
	aa := consul.New(l.data.consulClient)
	aa1, _ := aa.Watch(context.Background(), "private.comet")
	var zoneIns []*registry.ServiceInstance
	var err error
	for {
		if zoneIns, err = aa1.Next(); err != nil {
			break
		}
		for _, v := range zoneIns {
			l.nodes = append(l.nodes, v)
		}
		comets := map[string]comet.CometClient{}
		for _,v := range zoneIns{
			host := ""
			var grpcAddr string
			for _, addrs := range v.Endpoints {
				u, err := url.Parse(addrs)
				if err == nil && u.Scheme == "grpc" {
					grpcAddr = u.Host
					host = strings.Split(u.Host,":")[0]
				}
			}
			if old, ok := l.CometServers[host]; ok {
				comets[host] = old
				continue
			}
			c,err := l.NewCometClient(grpcAddr)
			if err != nil {
				continue
			}
			comets[host] = c
		}

		l.CometServers = comets
	}
}

func (l *LogicRepo) NewCometClient(addr string) (comet.CometClient,error)  {
	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(time.Second))
	defer cancel()
	conn, err := grpc.DialContext(ctx, addr,
		[]grpc.DialOption{
			grpc.WithInsecure(),
			grpc.WithInitialWindowSize(grpcInitialWindowSize),
			grpc.WithInitialConnWindowSize(grpcInitialConnWindowSize),
			grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(grpcMaxCallMsgSize)),
			grpc.WithDefaultCallOptions(grpc.MaxCallSendMsgSize(grpcMaxSendMsgSize)),
			grpc.WithKeepaliveParams(keepalive.ClientParameters{
				Time:                grpcKeepAliveTime,
				Timeout:             grpcKeepAliveTimeout,
				PermitWithoutStream: true,
			}),
		}...,
	)
	if err != nil {
		return nil, err
	}
	return comet.NewCometClient(conn), err
}